package baseFx.common.mq.mysql;

import baseFx.common.AutoCloseable;
import baseFx.common.Dynamic;
import baseFx.common.EventDispatcher;
import baseFx.common.concurrent.Task;
import baseFx.common.mq.MqMessage;
import baseFx.common.mq.MqSubscriber;
import baseFx.common.mq.SubscriberError;
import baseFx.common.utils.FxDate;
import baseFx.common.utils.StringUtils;
import baseFx.database.Sql;
import baseFx.database.Transaction;
import baseFx.database.TransactionLevel;
import baseFx.database.jdbc.impl.MysqlSql;

import javax.sql.DataSource;
import java.util.*;
import java.util.logging.Logger;

public class MysqlSubscriber implements MqSubscriber {
    private final static Logger logger = Logger.getLogger(MysqlSubscriber.class.getTypeName());
    private final static String msgkey = UUID.randomUUID().toString();
    private final static String errorkey = UUID.randomUUID().toString();
    private EventDispatcher<MqMessage> evMsg = new EventDispatcher<>(msgkey);
    private EventDispatcher<SubscriberError> evError = new EventDispatcher<>(errorkey);
    private static boolean isShutdown;
    private Map<String, QueueTask> tasks = new HashMap<>();
    private DataSource dataSource;

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> isShutdown = true));
    }

    public MysqlSubscriber(DataSource ds) {
        dataSource = ds;
    }

    protected Sql newSql() {
        return new MysqlSql(dataSource);
    }

    @Override
    public void declare(String queue) {

    }

    @Override
    public EventDispatcher<MqMessage> eventReceived() {
        return evMsg;
    }

    @Override
    public EventDispatcher<SubscriberError> eventError() {
        return evError;
    }

    private static String getKey(String queue, String name) {
        return queue + "$" + name;
    }

    @Override
    public void consume(String queue, String name, Object state) {
        String key = getKey(queue, name);
        QueueTask task = new QueueTask(queue, name, state);
        tasks.put(key, task);
        task.run();
    }

    @Override
    public void cancel(String queue, String name) {
        tasks.get(getKey(queue, name)).close();
    }

    protected void fireReceived(MqMessage message) {
        evMsg.fire(msgkey, this, message);
    }

    @Override
    public void close() {
        if (evMsg != null) {
            evMsg.clear();
            evMsg = null;
        }
        if (tasks != null) {
            for (QueueTask item : tasks.values()) {
                item.close();
            }
            tasks.clear();
            tasks = null;
        }
    }

    private class QueueTask implements AutoCloseable {
        private Sql sql;
        private Object state;
        private String queue;
        private String name;
        private boolean running;
        private String queueId;

        public QueueTask(String queue, String name, Object state) {
            this.state = state;
            this.queue = queue;
            this.name = name;
        }

        public void run() {
            this.sql = newSql();
            try (Transaction tx = sql.transaction(TransactionLevel.Serializable)) {
                queueId = sql.queryOne(String.class, "select id from mq_queue where `queue`=#q and `name`=#n", Dynamic.setNew(s -> {
                    s.put("q", queue);
                    s.put("n", name);
                }));
                if (queueId == null) {
                    queueId = UUID.randomUUID().toString();
                    sql.execute("insert into mq_queue values(#id,#q,#n)", Dynamic.setNew(s -> {
                        s.put("id", queueId);
                        s.put("q", queue);
                        s.put("n", name);
                    }));
                }
                tx.commit();
            }
            running = true;
            startTask();
        }

        private void startTask() {
            Task.run(() -> {
                String sqlTmp = "select id,msg,re_count from mq_msg where queue_id='" + queueId + "' and next_time<{} limit 500 for update";
                while (running) {
                    Transaction tx = sql.transaction();
                    try {
                        List<Dynamic> valid = sql.query(StringUtils.format(sqlTmp, new FxDate().yyyyMMddHHmmss()));
                        if (valid.size() == 0) {
                            tx.rollback();
                            Task.delay(1000, 2000);
                            continue;
                        }
                        List<String> del = new ArrayList<>();
                        List<Dynamic> re = new ArrayList<>();
                        for (Dynamic item : valid) {
                            MqMessage message = new MqMessage();
                            message.name = name;
                            message.queue = queue;
                            message.id = item.getString("id");
                            message.state = state;
                            message.data = item.getString("msg");
                            fireReceived(message);
                            if (message.reconsume) {
                                re.add(item);
                            } else {
                                del.add(message.id);
                            }
                        }
                        if (del.size() > 0) {
                            sql.execute("delete from mq_msg where id in #ids", Dynamic.setNew(s -> s.put("ids", del)));
                        }
                        for (Dynamic item : re) {
                            int rc = item.getInt("re_count") + 1;
                            sql.execute("update mq_msg set is_hold=0,next_time=#time,re_count=#rc where id=#id", Dynamic.setNew(s -> {
                                s.put("id", item.getString("id"));
                                s.put("time", new FxDate().addMinute(rc).yyyyMMddHHmmss());
                                s.put("rc", rc);
                            }));
                        }
                        tx.commit();
                    } catch (Exception e) {
                        SubscriberError args = new SubscriberError(e, state);
                        evError.fire(errorkey, MysqlSubscriber.this, args);
                        if (args.isRetry == false || isShutdown) {
                            throw e;
                        }
                    }
                }
            }).onCompleted(t -> {
                if (t.getError() != null && isShutdown == false) {
                    logger.severe(t.getError().getClass().getSimpleName() + ":" + t.getError().getMessage());
                }
                if (sql != null) {
                    sql.close();
                }
                MysqlSubscriber.this.tasks.remove(getKey(queue, name));
            });
        }

        @Override
        public void close() {
            running = false;
        }
    }
}
